#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>
#include <netdb.h>
#include <sys/types.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include<string.h>
#include <unistd.h>
#include <errno.h>
#include<time.h>
#include<pthread.h>
#include<sys/time.h>
#include<semaphore.h>
#define SERV_PORT 8888//服务端口号
#define MAXSIZE 1000//缓存队列可以存储的最大数据
#define BUF_COUNT 2//缓存去的个数
#define MSG_FILENAME 1
#define MSG_CONTINUE 2
#define MSG_ACK 3
#define MSG_DONE 4
#define MSG_EXCEPTION 5
//消息结构体
struct msg
{
    int type;
    int len;
    char data[];
};
sem_t count;//已经使用的缓冲区
sem_t empty_count;//未使用的缓冲区
int my_count;
int head;//队头指针
int rear;//队尾指针
struct msg *buf_ptr[BUF_COUNT];//缓冲区队列
//将消息结构体，拷贝进缓冲区
void cpy_sm_to_buf(struct msg *sm)
{
    sem_wait(&empty_count);
    memset(buf_ptr[rear],0x00,MAXSIZE);
    memcpy(buf_ptr[rear],sm,sizeof(struct msg)+sm->len);
    //更新队列状态
    rear = (rear+1)%BUF_COUNT;
    sem_post(&count);
}
//生产者，读取文件并放入缓存区
void* producer(void *arg)
{
    char *filepath=(char *)arg;
    FILE *fp;
    int i,j;
    int thread_exit_flag=0;
    int datalen;
    int produce_flag;
    struct msg *sm;
    sm = (struct msg*)malloc(MAXSIZE+sizeof(struct msg));
    //由文件路径获取文件名
    char file_name[256];
    memset(file_name,0x00,256);
    for(i=strlen(filepath)-1; i>=0; i--)
    {
        if(filepath[i] == '/')
        {
            break;
        }
    }
    for(j=i+1; j<strlen(filepath); j++)
    {
        file_name[j-i-1]=filepath[j];
    }
    sm->type = MSG_FILENAME;
    sm->len = strlen(file_name);
    memset(sm->data,0x00,MAXSIZE);
    memcpy(sm->data,file_name,sm->len);//将文件名拷贝到消息中
    cpy_sm_to_buf(sm);
//打开文件
    if((fp=fopen(filepath,"r")) == NULL)
    {
        perror("fopen error");
        exit(-1);
    }
//循环读取文件
    produce_flag = MAXSIZE;
    while(produce_flag == MAXSIZE)
    {
        sm->type = MSG_CONTINUE;
        memset(sm->data,0x00,MAXSIZE);
        if((datalen=fread(sm->data,1,MAXSIZE,fp)) == 0)
        {
            if(feof(fp))
            {
                printf("注意，文件为空！\n");
            }
            else
            {
                sm->type = MSG_EXCEPTION;
                sm->len = 0;
                memset(sm->data,0x00,MAXSIZE);
                cpy_sm_to_buf(sm);
                produce_flag = datalen;
                printf("fread error\n");
                exit(-1);
            }
        }
        sm->len = datalen;
        cpy_sm_to_buf(sm);
        produce_flag = datalen;
    }
    sm->type = MSG_DONE;
    sm->len = 0;
    memset(sm->data,0x00,MAXSIZE);
    cpy_sm_to_buf(sm);
    fclose(fp);
    free(sm);
    pthread_exit(NULL);
}
//消费者，将缓存区的内容发送到服务端
void* consumer(void *arg)
{
    int consufd = *(int *)arg;
    int i;
    int consume_flag = MAXSIZE;
    while(consume_flag != 0)
    {
        sem_wait(&count);
        //向服务端发送文件名消息
        consume_flag = buf_ptr[head]->len;
        if((send(consufd,buf_ptr[head],sizeof(struct msg)+buf_ptr[head]->len,0)) == -1)
        {
            perror("send error");
            exit(-1);
        }
        //更新队列状态
        head = (head+1)%BUF_COUNT;
        sem_post(&empty_count);
    }
    pthread_exit(NULL);
}
int main(int argc,char *argv[])
{
    if(argc!=3)
    {
        printf("%s: input IP & filemane\n",argv[0]);
        return 1;
    }
    int i;
    head = 0;
    rear = 0;
    struct sockaddr_in their_addr;
    pthread_t ptid1,ptid2;
    int sockfd,numbytes;
    struct timeval start;
    struct timeval end;
    struct hostent *he;
    struct msg *rm;
    rm = (struct msg*)malloc(sizeof(struct msg)+MAXSIZE);
    //为存放数据的缓冲去分配内存
    for(i=0; i<BUF_COUNT; i++)
    {
        buf_ptr[i]=(struct msg *)malloc(MAXSIZE+sizeof(struct msg));
    }
    //将基本名字和地址转换
    he = gethostbyname(argv[1]);
    //建立一个TCP套接口
    if((sockfd = socket(AF_INET,SOCK_STREAM,0))==-1)
    {
        perror("socket");
        exit(1);
    }
    //初始化结构体
    their_addr.sin_family = AF_INET;
    their_addr.sin_port = htons(SERV_PORT);
    their_addr.sin_addr =*((struct in_addr *)he->h_addr);
    bzero(&(their_addr.sin_zero),8);
    //和服务端建立连接
    if(connect(sockfd,(struct sockaddr *)&their_addr,sizeof(struct sockaddr))==-1)
    {
        perror("connect");
        printf("%s\n",strerror(errno));
        exit(1);
    }
    if(sem_init(&count,0,0) == -1)
    {
        perror("sem_init error");
        exit(-1);
    }
    if(sem_init(&empty_count,0,BUF_COUNT) == -1)
    {
        perror("sem_init error");
        exit(-1);
    }
    gettimeofday(&start,NULL);
    pthread_create(&ptid1,NULL,producer,argv[2]);
    pthread_create(&ptid2,NULL,consumer,&sockfd);
    pthread_join(ptid1,NULL);
    pthread_join(ptid2,NULL);
    gettimeofday(&end,NULL);
    sem_destroy(&count);
    sem_destroy(&empty_count);
    printf("the time of sending the file is %ld\n",(end.tv_sec - start.tv_sec)*1000000+(end.tv_usec - start.tv_usec));
    //接受服务端发送的确认消息
    if((numbytes = recv(sockfd,(void *)rm,sizeof(struct msg),0)) == -1)
    {
        perror("recv error");
        exit(-1);
    }
    printf("send MSG_DONE!\n");
    close(sockfd);
    //释放缓冲队列
    for(i=0; i<BUF_COUNT; i++)
    {
        free(buf_ptr[i]);
    }
    free(rm);
    return 0;
}
